package com.sparkstream;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.StreamingContext;

import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.junit.Test;
import scala.Tuple2;

import java.io.Serializable;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;

public class TestSparkStreaming implements Serializable {

    Map<String, Long> map = new HashMap<>();
    @Test
    public void test() throws InterruptedException {
        SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount");

        JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(5));

        JavaReceiverInputDStream<String> lines = jssc.socketTextStream("master", 9999);
        lines.print();

        //
        JavaDStream<String> dStream = lines.flatMap(t -> Arrays.asList(t.split(" ")).iterator());
        JavaPairDStream<String, Integer> pairDStream = dStream.mapToPair(t -> new Tuple2<>(t, 1));
        pairDStream.print();
//        pairDStream.countByValue();
        JavaPairDStream<String, Long> stringLongJavaPairDStream = dStream.countByValue();

        stringLongJavaPairDStream.foreachRDD(t -> {
            t.foreach(t1 -> {
                map.put(t1._1, map.get(t) + 1);
            });
        });
        stringLongJavaPairDStream.print();

        //切分
//        JavaDStream<String> map = lines.map(new Function<String, String>() {
//            @Override
//            public String call(String s) throws Exception {
//                return "结果为：" + s;
//            }
//        });
//        map.print();

//        lines.flatMap(new FlatMapFunction<String, Object>() {
//            @Override
//            public Iterator<Object> call(String s) throws Exception {
//                return null;
//            }
//        });

//        lines.filter(new Function<String, Boolean>() {
//            @Override
//            public Boolean call(String s) throws Exception {
//                return null;
//            }
//        });

//        lines.mapToPair(new PairFunction<String, Object, Object>() {
//            @Override
//            public Tuple2<Object, Object> call(String s) throws Exception {
//                return null;
//            }
//        });

        jssc.start();              // Start the computation
        jssc.awaitTermination();   // Wait for the computation to terminate
    }
}
